package simplifiedPastry;

import java.io.IOException;
import java.math.BigInteger;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.net.UnknownHostException;
import java.security.InvalidParameterException;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.Iterator;
import java.util.Map;
import java.util.Random;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.Map.Entry;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import simplifiedPastry.exceptions.NodeConnectionException;
import simplifiedPastry.exceptions.NodeConnectionIOErrorException;
import simplifiedPastry.exceptions.NodeConnectionTimeOutException;
import simplifiedPastry.exceptions.NodeStateException;

/**
 * Class representing Node in simplified-pastry peer-to-peer network library.
 * 
 * @author Stanisław Ogórkis
 */
public class Node {
	
	final private static int msgListnEventsListSize = 100;
	final private static long rtRequestSendTime = 1 * 60 * 1000L;
	
	protected NodeId		nodeId;
	private InetAddress		inetAddress;
	private int				port;
	private NodeEntry		nodeEntry;
	private DatagramSocket	recvSocket, sendSocket;
	private RoutingTable	routingTable;
	private LeafSet			leafSet;
	private MessageListener messageListener;
	private ExecutorService	executor;
	private Boolean			isConnected, conn;
	private long			connectTimeout;
	private long			sendWaitTimeout;
	private Random			randomGen;
	private long			rtRequestTime;
	private int				rtRespSize;
	private InetSocketAddress							inetSocketAddress;
	private SocketListenerRunnable						socketListenerRunnable;
	private Map<Long, MessageWaitEntry>					msgTimeOutMap;
	private ArrayBlockingQueue<MessageListenerEvent>	msgListnEventsList;
	
	public Node(NodeId nodeId, int port, long connectTimeout, long sendWaitTimeout) {
		this.nodeId			= nodeId;
		this.port			= port;
		this.connectTimeout	= connectTimeout;
		this.sendWaitTimeout= sendWaitTimeout;
		nodeEntry			= new NodeEntry(nodeId, inetAddress, port);
		routingTable		= new RoutingTable(nodeEntry);
		leafSet				= new LeafSet(nodeEntry);
		randomGen			= new Random();
		executor			= Executors.newFixedThreadPool(2);
		isConnected = conn	= Boolean.FALSE;
		msgListnEventsList	= new ArrayBlockingQueue<MessageListenerEvent>(msgListnEventsListSize);
		msgTimeOutMap		= Collections.synchronizedMap(new TreeMap<Long, MessageWaitEntry>());
		try {
			inetAddress			= InetAddress.getLocalHost();
			inetSocketAddress	= new InetSocketAddress(inetAddress, port);
		} catch (UnknownHostException e) { }
	}
	
	public Node(NodeId nodeId, InetAddress inetAddress, int port, long connectTimeout
			, long sendWaitTimeout) {
		this(nodeId, port, connectTimeout, sendWaitTimeout);
		this.inetAddress	= inetAddress;
		nodeEntry.inetAddress = inetAddress;
		routingTable		= new RoutingTable(nodeEntry);
		leafSet				= new LeafSet(nodeEntry);
		inetSocketAddress	= new InetSocketAddress(inetAddress, port);
	}
	
	public Node(NodeId nodeId, InetAddress inetAddress, int port, long connectTimeout,
			long sendWaitTimeout, MessageListener messageListener) {
		this(nodeId, inetAddress, port, connectTimeout, sendWaitTimeout);
		this.messageListener= messageListener;
	}
	
	protected void setMessageListener(MessageListener messageListener) throws NodeStateException {
		if (this.messageListener != null) {
			throw new NodeStateException();
		}
		this.messageListener= messageListener;		
	}
	
	public void connect(InetAddress address, int port)
	throws NodeStateException, NodeConnectionException, SocketException {
		if (isConnected) {
			throw new NodeStateException();
		}
		recvSocket = new DatagramSocket(this.port, inetAddress);
		sendSocket = new DatagramSocket();
		recvSocket.setSoTimeout(1000);
		executor.execute(new MessageListnerRunnable());
		socketListenerRunnable = new SocketListenerRunnable(recvSocket);
		executor.execute(socketListenerRunnable);
		try {
			processJoin(inetAddress, port);
		} catch (NodeConnectionException e) {
			executor.shutdownNow();
			throw e;
		}
		isConnected = true;
	}
	
	public void createNewNet() throws SocketException, NodeStateException {
		if (isConnected) {
			throw new NodeStateException();
		}
		recvSocket = new DatagramSocket(port, inetAddress);
		sendSocket = new DatagramSocket();
		recvSocket.setSoTimeout(1000);
		executor.execute(new MessageListnerRunnable());
		socketListenerRunnable = new SocketListenerRunnable(recvSocket);
		executor.execute(socketListenerRunnable);
		isConnected = true;
	}
	
	public void disconnect() {
		processDisconnect();
		executor.shutdown();
		try {
			executor.awaitTermination(1, TimeUnit.SECONDS);
		} catch (InterruptedException e) { }
		executor.shutdownNow();
	}
	
	public void sendMessage(Message message, NodeId recvId) {
		NodeEntry nodeEntry = nextHop(recvId);
		if (nodeEntry == null) {
			if(nodeId.compareTo(recvId) == 0) {
				msgListnEventsList.offer(new MessageRecvEvent(nodeId, message.getByte()));
			} else {
				msgListnEventsList.offer(new NodeNotFoundEvent(recvId));
			}
			return ;
		}
		byte[] buff = new byte[44+message.getByte().length];
		uShortToBuff(buff, 0, PacketHeaders.MESSAGE);
		fillByteArrayBuff(buff, 2, recvId.getByteNodeId());
		fillByteArrayBuff(buff, 18, nodeId.getByteNodeId());
		fillByteArrayBuff(buff, 34, inetAddress.getAddress());
		uShortToBuff(buff, 38, port);
		int randVal = randomGen.nextInt();
		randVal = randVal >>> 1;
		while (msgTimeOutMap.containsKey(randVal)) {
			randVal = randomGen.nextInt();
			randVal = randVal >>> 1;
		}
		uIntToBuff(buff, 40, randVal);
		fillByteArrayBuff(buff, 44, message.getByte());
		DatagramPacket kSearchPacket = new DatagramPacket(buff, buff.length,
				nodeEntry.inetAddress, nodeEntry.port);
		synchronized (sendSocket) {
			try {
				sendSocket.send(kSearchPacket);
				msgTimeOutMap.put(new Long(randVal), new MessageWaitEntry(message, randVal));
			} catch (IOException e) {
				msgListnEventsList.offer(new IOErrorEvent(e));
			}
		}
	}
	
	public void kSearch(NodeId searchId, int num) {
		kSearch(searchId, num, inetSocketAddress);
	}
	
	private NodeEntry nextHop(NodeId hopNodeId) {
		if (hopNodeId.compareTo(nodeId) == 0) {
			return null;
		}
		NodeEntry nodeEntry = null;
		if ((nodeEntry = leafSet.getNextNodeEntry(hopNodeId)) != null) {
			System.out.println("found - leafSet!");
			return nodeEntry;
		} else if ((nodeEntry = routingTable.getNextNodeEntry(hopNodeId)) != null) {
			System.out.println("found - routingTable!");
			return nodeEntry;
		}
		return null;
	}
	
	private void processNodeNotFound(NodeId notFoundNode, NodeId recvId) {
		byte[] buff = new byte[36];
		uShortToBuff(buff, 0, PacketHeaders.ERROR);
		fillByteArrayBuff(buff, 2, recvId.getByteNodeId());
		fillByteArrayBuff(buff, 20, notFoundNode.getByteNodeId());
		uIntToBuff(buff, 18, PacketHeaders.ERROR_NODE_NOT_FOUND);
		NodeEntry nodeEntry = nextHop(recvId);
		if (nodeEntry == null) {
			return ;
		}
		DatagramPacket packet = new DatagramPacket(buff, buff.length); 
		packet.setAddress(nodeEntry.inetAddress);
		packet.setPort(nodeEntry.port);
		synchronized (sendSocket) {
			try {
				sendSocket.send(packet);
			} catch (IOException e) {
				msgListnEventsList.offer(new IOErrorEvent(e));
			}
		}
	}
	
	private void kSearch(NodeId searchId, int num, InetSocketAddress addr) {
		if (num > ProtocolVariables.leafSetSize || searchId.compareTo(nodeId) == 0) {
			throw new InvalidParameterException();
		}
		NodeEntry nodeEntry = nextHop(searchId);
		if (nodeEntry != null) {
			System.out.println("kSearch nextHop: " + nodeEntry.nodeId
					+ " " + nodeEntry.inetAddress + " " + nodeEntry.port);
		} else {
			System.out.println("kSearch nextHop: null");
		}
		if (nodeEntry == null) {
			processKSearchResponse(searchId, ProtocolVariables.leafSetSize, addr);
			return ;
		}
		byte[] buff = new byte[26];
		uShortToBuff(buff, 0, PacketHeaders.KSEARCH);
		fillByteArrayBuff(buff, 2, searchId.getByteNodeId());
		fillByteArrayBuff(buff, 18, addr.getAddress().getAddress());
		uShortToBuff(buff, 22, addr.getPort());
		uShortToBuff(buff, 24, num);
		DatagramPacket kSearchPacket = new DatagramPacket(buff, buff.length,
				nodeEntry.inetAddress, nodeEntry.port);
		synchronized (sendSocket) {
			try {
				sendSocket.send(kSearchPacket);
			} catch (IOException e) {
				msgListnEventsList.offer(new IOErrorEvent(e));
			}
		}
	}
	
	private void processJoin(InetAddress address, int port)
	throws NodeConnectionException {
		byte[] buff = new byte[24];
		uShortToBuff(buff, 0, PacketHeaders.JOIN_REQUEST);
		fillByteArrayBuff(buff, 2, nodeId.getByteNodeId());
		fillByteArrayBuff(buff, 18, inetAddress.getAddress());
		uShortToBuff(buff, 22, inetSocketAddress.getPort());
		DatagramPacket joinPacket = new DatagramPacket(buff, buff.length, address, port);
		synchronized (sendSocket) {
			try {
				for(int i = 0; i < 3; ++i) {
					sendSocket.send(joinPacket);
					synchronized (isConnected) {
						isConnected.wait(connectTimeout/3);
						if (conn == true) {
							isConnected = true;
							break;
						}
					}
				}
				if (isConnected == false) {
					throw new NodeConnectionTimeOutException();
				}
			} catch (IOException e) {
				throw new NodeConnectionIOErrorException();
			} catch (InterruptedException e) { }
		}
		isConnected = false;
		rtRespSize = processRTRequest();
		synchronized (isConnected) {
			try {
				isConnected.wait(connectTimeout);
			} catch (InterruptedException e) { }
			isConnected = true;
		}
		NodeEntry[] leafSetEntries = leafSet.getKLeafSet(ProtocolVariables.leafSetSize);
		NodeEntry[] rtEntries = routingTable.getRTEntries();
		System.out.println("Almost joined: leafSet");
		System.out.println(leafSet);
		System.out.println();
		for(int i = 0; i < leafSetEntries.length; ++i) {
			processRTRequestResponse(leafSetEntries[i]);
		}
		for(int i = 0; i < rtEntries.length; ++i) {
			processRTRequestResponse(rtEntries[i]);
		}
	}
	
	private int processRTRequest() {
		rtRequestTime = System.currentTimeMillis();
		NodeEntry[] leafSetEnt	= leafSet.getLeafSet();
		if (leafSetEnt.length == 0) {
			return -1;
		}
		byte[] buff = new byte[24];
		uShortToBuff(buff, 0, PacketHeaders.RT_REQUEST);
		fillByteArrayBuff(buff, 2, nodeId.getByteNodeId());
		fillByteArrayBuff(buff, 18, inetAddress.getAddress());
		uShortToBuff(buff, 22, port);
		DatagramPacket packet = new DatagramPacket(buff, buff.length);
		for(int i = 0; i < leafSetEnt.length; ++i) {
			packet.setAddress(leafSetEnt[i].inetAddress);
			packet.setPort(leafSetEnt[i].port);
			synchronized (sendSocket) {
				try {
					sendSocket.send(packet);
				} catch (IOException e) {
					msgListnEventsList.offer(new IOErrorEvent(e));
				}
			}
		}
		return leafSetEnt.length;
	}

	private void processDisconnect() {
		byte[] buff;
		NodeEntry[] leafSetEnt	= leafSet.getLeafSet();
		synchronized (routingTable) {
			int rtSize = routingTable.getSize()*22;
			buff = new byte[358+rtSize];
			routingTable.fillRoutingTableBuff(buff, 358);
			uIntToBuff(buff, 354, rtSize);
		}
		uShortToBuff(buff, 0, PacketHeaders.RT_UPDATE);
		int offset = 2;
		for (int i = 0; i < leafSetEnt.length; ++i) {
			leafSetEnt[i].fillByteBuff(buff, offset);
			offset += 22;
		}
		DatagramPacket packet = new DatagramPacket(buff, buff.length);
		for(int i = 0; i < leafSetEnt.length; ++i) {
			packet.setAddress(leafSetEnt[i].inetAddress);
			packet.setPort(leafSetEnt[i].port);
			synchronized (sendSocket) {
				try {
					sendSocket.send(packet);
				} catch (IOException e) {
					msgListnEventsList.offer(new IOErrorEvent(e));
				}
			}
		}
	}
	
	private void processKSearchResponse(NodeId searchId, int num, InetSocketAddress addr) {
		int leafSetSize, buffSize, sendNum;
		byte[] buff = null;
		synchronized (leafSet) {
			leafSetSize = leafSet.getSize();
			buffSize = (leafSetSize < num) ? leafSetSize : num;
			sendNum = buffSize;
			if (buffSize == 0) {
				buffSize = 1;
			}
			buffSize *= 22;
			buffSize += 36;
			buff = new byte[buffSize];
			leafSet.fillLeafSetByteData(buff, 36);
		}
		uShortToBuff(buff, 0, PacketHeaders.KSEARCH_RESPONSE);
		fillByteArrayBuff(buff, 2, searchId.getByteNodeId());
		fillByteArrayBuff(buff, 18, nodeId.getByteNodeId());
		if (sendNum == 0) {
			fillByteArrayBuff(buff, 36, nodeId.getByteNodeId());
			fillByteArrayBuff(buff, 52, inetAddress.getAddress());
			uShortToBuff(buff, 56, inetSocketAddress.getPort());
			++sendNum;
		}
		uShortToBuff(buff, 34, sendNum);
		DatagramPacket kSearchRespPacket = new DatagramPacket(buff, buff.length,
				addr.getAddress(), addr.getPort());
		synchronized (sendSocket) {
			try {
				sendSocket.send(kSearchRespPacket);
			} catch (IOException e) {
				msgListnEventsList.offer(new IOErrorEvent(e));
			}
		}
	}

	private void processRTRequestResponse(NodeEntry sendNode) {
		byte[] buff;
		NodeEntry[] leafSetEnt	= leafSet.getLeafSet();
		synchronized (routingTable) {
			int rtSize = routingTable.getSize()*22;
			buff = new byte[380+rtSize];
			routingTable.fillRoutingTableBuff(buff, 380);
			uIntToBuff(buff, 376, rtSize);
		}
		nodeEntry.fillByteBuff(buff, 2);
		uShortToBuff(buff, 0, PacketHeaders.RT_UPDATE);
		int offset = 24;
		for (int i = 0; i < leafSetEnt.length; ++i) {
			leafSetEnt[i].fillByteBuff(buff, offset);
			offset += 22;
		}
		DatagramPacket packet = new DatagramPacket(buff, buff.length);
		packet.setAddress(sendNode.inetAddress);
		packet.setPort(sendNode.port);
		synchronized (sendSocket) {
			try {
				sendSocket.send(packet);
			} catch (IOException e) {
				msgListnEventsList.offer(new IOErrorEvent(e));
			}
		}
	}
	
	private void fillByteArrayBuff(byte[] buff, int offset, byte[] byteArray) {
		for(int i = 0; i < byteArray.length && i+offset < buff.length; ++i) {
			buff[i+offset] = byteArray[i];
		}
	}
	
	private void uShortToBuff(byte[] buff, int offset, int ushortVal) {
		buff[offset]	= (byte) ((ushortVal & 0xFF00) >> 8);
		buff[offset+1]	= (byte) (ushortVal & 0x00FF);
	}
	
	private void uIntToBuff(byte[] buff, int offset, long uintVal) {
		buff[offset]	= (byte) ((uintVal & 0xFF000000) >> 24);
		buff[offset+1]	= (byte) ((uintVal & 0x00FF0000) >> 16);
		buff[offset+2]	= (byte) ((uintVal & 0x0000FF00) >> 8);
		buff[offset+3]	= (byte) (uintVal & 0x000000FF);
	}
	
	private long buffToUInt(byte[] buff, int offset) {
		long l = 0;
	    l |= buff[offset] & 0xFF;
	    l <<= 8;
	    l |= buff[offset+1] & 0xFF;
	    l <<= 8;
	    l |= buff[offset+2] & 0xFF;
	    l <<= 8;
	    l |= buff[offset+3] & 0xFF;
	    return l;

	}
	
	private int buffToUShort(byte[] buff, int offset) {
		int i = 0;
	    i |= buff[offset+0] & 0xFF;
	    i <<= 8;
	    i |= buff[offset+1] & 0xFF;
	    return i;

	}
	
	private class MessageListenerEvent { }
	
	private class KSearchResponseEvent extends MessageListenerEvent {
		
		private NodeId		searchNodeId;
		private NodeId[]	nodeIdArray;
		
		public KSearchResponseEvent(NodeId searchNodeId, NodeId[] nodeIdArray) {
			this.searchNodeId= searchNodeId;
			this.nodeIdArray = nodeIdArray;
		}
		
		public NodeId[] getNodeIdArray() {
			return nodeIdArray;
		}
		
		public NodeId getSearchNodeId() {
			return searchNodeId;
		}
	}
	
	private class MessageRecvEvent extends MessageListenerEvent {
		
		private NodeId	senderId;
		private byte[]	msgBuff;
		
		public MessageRecvEvent(NodeId senderId, byte[] msgBuff) {
			this.senderId	= senderId;
			this.msgBuff	= msgBuff;
		}
		
		public NodeId getSenderId() {
			return senderId;
		}

		public byte[] getMsgBuff() {
			return msgBuff;
		}

	}
	
	private class MessageAcceptedEvent extends MessageListenerEvent {
		
		private NodeId	senderId;
		private long	msgId;
		
		public MessageAcceptedEvent(NodeId senderId, long msgId) {
			this.senderId	= senderId;
			this.msgId		= msgId;
		}
		
		public NodeId getSenderId() {
			return senderId;
		}

		public long getMsgId() {
			return msgId;
		}

	}
	
	private class NodeNotFoundEvent extends MessageListenerEvent {
		
		private NodeId	errorNodeId;
		
		public NodeNotFoundEvent(NodeId errorNodeId) {
			this.errorNodeId	= errorNodeId;
		}
		
		public NodeId getErrorNodeId() {
			return errorNodeId;
		}

	}
	
	private class IOErrorEvent extends MessageListenerEvent {
		
		private IOException	exception;
		
		public IOErrorEvent(IOException exception) {
			this.exception	= exception;
		}
		
		public IOException getErrorException() {
			return exception;
		}

	}
	
	private class MessageTimeOutEvent extends MessageListenerEvent {
		
		private Message msg;
		
		public MessageTimeOutEvent(Message msg) {
			this.msg	= msg;
		}
		
		public Message getErrorException() {
			return msg;
		}

	}
	
	private class MessageWaitEntry {
		
		public Message	msg;
		public long		msgId;
		private long	sendTime;
		
		public MessageWaitEntry(Message msg, long msgId) {
			this.msg		= msg;
			this.msgId		= msgId;
			this.sendTime	= System.currentTimeMillis();
		}
		
		public boolean checkExpired() {
			return (System.currentTimeMillis() - sendTime > sendWaitTimeout);
		}
	}
	
	public class NodeEntry {
		
		public NodeId nodeId;
		public InetAddress inetAddress;
		public int port;
		
		public NodeEntry(NodeId nodeId, InetAddress address, int port) {
			this.nodeId		= nodeId;
			this.inetAddress= address;
			this.port		= port;
		}

		public NodeEntry() {
		}

		public void fillByteBuff(byte[] buff, int offset) {
			fillByteArrayBuff(buff, offset, nodeId.getByteNodeId());
			fillByteArrayBuff(buff, offset+16, inetAddress.getAddress());
			uShortToBuff(buff, offset+20, port);
		}
		
		@Override
		public boolean equals(Object obj) {
			if (obj instanceof NodeEntry) {
				NodeEntry entry = (NodeEntry) obj;
				if (entry.nodeId.equals(nodeId) && entry.port == port
						&& entry.inetAddress.equals(inetAddress)) {
					return true;
				}
			}
			return false;
		}
	}	
	
	private class MessageListnerRunnable implements Runnable {
		
		@Override
		public void run() {
			Thread.currentThread().setName("MessageListenerThread");
			while(true) {
				MessageListenerEvent event;
				try {
					event = msgListnEventsList.take();
				} catch (InterruptedException e) {
					break ;
				}
				if (event instanceof KSearchResponseEvent) {
					KSearchResponseEvent kSerchEvent = (KSearchResponseEvent) event;
					messageListener.kSearchRecv(kSerchEvent.getNodeIdArray());
				} else if (event instanceof MessageAcceptedEvent) {
					MessageAcceptedEvent msgAcceptEvent = (MessageAcceptedEvent) event;
					MessageWaitEntry msgWaitEntry = msgTimeOutMap.remove(msgAcceptEvent.getMsgId());
					if (msgWaitEntry == null) {
						return ;
					}
					Message msg = msgWaitEntry.msg;
					messageListener.messageAccepted(msg);
				} else if (event instanceof MessageRecvEvent) {
					MessageRecvEvent msgRecvEvent = (MessageRecvEvent) event;
					messageListener.messageRecv(new Message(msgRecvEvent.getMsgBuff()),
							msgRecvEvent.getSenderId());
				} else if (event instanceof NodeNotFoundEvent) {
					NodeNotFoundEvent notFoundEvent = (NodeNotFoundEvent) event;
					messageListener.nodeNotFound(notFoundEvent.getErrorNodeId());
				} else if (event instanceof IOErrorEvent) {
					IOErrorEvent errorEvent = (IOErrorEvent) event;
					messageListener.ioError(errorEvent.getErrorException());
				} else if (event instanceof MessageTimeOutEvent) {
					MessageTimeOutEvent msgTimeOutEvent = (MessageTimeOutEvent) event;
					messageListener.messageTimeout(msgTimeOutEvent.msg);
				}
			}		
		}
	}
	
	private class SocketListenerRunnable implements Runnable {
		
		private boolean isJoined;
		private DatagramSocket	socket;
		final private static int buffSize = 1024*80;
		
		public SocketListenerRunnable(DatagramSocket socket) {
			this.socket	= socket;
		}

		@Override
		public void run() {
			int joinNumWait = 0;
			Thread.currentThread().setName("SocketListenerThread");
			byte[] buff = new byte[buffSize];
			DatagramPacket recvPacket = new DatagramPacket(buff, buff.length);
			while(true) {
				try {
					socket.receive(recvPacket);
					if (Thread.currentThread().isInterrupted()) {
						return ;
					}
				} catch (SocketTimeoutException e) {
					if (Thread.currentThread().isInterrupted()) {
						return ;
					}
					processMsgWaitUpdate();
					continue;
				} catch (IOException e) {
					msgListnEventsList.offer(new IOErrorEvent(e));
					continue ;
				}
				byte[] packetData = recvPacket.getData();
				if (packetData.length < 24) {
					continue;
				}
				int packetHeader = buffToUShort(buff, 0);

				switch (packetHeader) {
				case PacketHeaders.JOIN_REQUEST:
				{
					if(recvPacket.getLength() != 24) {
						break ;
					}
					byte[] ipAddress		= Arrays.copyOfRange(buff, 18, 22);
					int port				= buffToUShort(buff, 22);
					InetSocketAddress addr = null;
					try {
						addr = new InetSocketAddress(
								InetAddress.getByAddress(ipAddress), port);
						kSearch(new NodeId(buff, 2), ProtocolVariables.leafSetSize, addr);
						NodeEntry entry = new NodeEntry(new NodeId(buff, 2),
								addr.getAddress(),addr.getPort());
						routingTable.addNodeEntry(entry);
						leafSet.addNodeEntry(entry);
					} catch (UnknownHostException e) { e.printStackTrace(); }	
					break;
				}
				case PacketHeaders.KSEARCH:
				{
					if(recvPacket.getLength() != 26) {
						break ;
					}
					NodeId kSearchNodeId	= new NodeId(buff, 2);
					NodeEntry entry			= null;
					if((entry = nextHop(kSearchNodeId)) == null) {
						byte[] ipAddress	= Arrays.copyOfRange(buff, 18, 22);
						int port			= buffToUShort(buff, 22);
						int kSize			= buffToUShort(buff, 24);
						processKsearchResponse(kSearchNodeId, ipAddress, port, kSize);
						break;
					}
					forwardKSearch(entry, buff);
					break;
				}
				case PacketHeaders.KSEARCH_RESPONSE:
				{
					if(recvPacket.getLength() < 36) {
						break ;
					}
					NodeId kSearchId = new NodeId(buff, 2);
					int kSize = buffToUShort(buff, 34);
					NodeId[] nodeIdArray = new NodeId[kSize];
					for(int i = 0; i < kSize; ++i) {
						nodeIdArray[i]	= new NodeId(buff, i*22+36);
						NodeEntry entry = new NodeEntry();
						entry.nodeId	= nodeIdArray[i];
						byte[] ipAddress= Arrays.copyOfRange(buff, i*22+52, i*22+56);
						try {
							entry.inetAddress = InetAddress.getByAddress(ipAddress);
						} catch (UnknownHostException e) {
							e.printStackTrace();
							continue;
						}
						entry.port		= buffToUShort(buff, i*22+56);
						routingTable.addNodeEntry(entry);
						leafSet.addNodeEntry(entry);
					}
					if (isJoined == false) {
						isJoined = true;
						synchronized (isConnected) {
							conn = true;
							isConnected.notify();
						}
						break;
					}
					msgListnEventsList.offer(new KSearchResponseEvent(kSearchId, nodeIdArray));
					break;
				}
				case PacketHeaders.MESSAGE:
				{
					if(recvPacket.getLength() < 44) {
						break ;
					}
					NodeId recvNodeId	= new NodeId(buff, 2);
					if (nodeId.compareTo(recvNodeId) != 0) {
						forwardSendMessage(recvNodeId, buff, recvPacket.getLength());
						break;
					}
					msgListnEventsList.offer(new MessageRecvEvent(new NodeId(
							buff, 18), Arrays.copyOfRange(buff, 44,
									recvPacket.getLength())));
					long msgId = buffToUInt(buff, 40);
					byte[] ipAddress= Arrays.copyOfRange(buff, 34, 38);
					int port = buffToUShort(buff, 38);
					try {
						processMessageAcceptResp(new NodeId(buff, 18),
								InetAddress.getByAddress(ipAddress),
								port ,msgId);
					} catch (UnknownHostException e) {
						e.printStackTrace();
					}
					break;
				}
				case PacketHeaders.MESSAGE_ACCEPTED:
				{
					if(recvPacket.getLength() != 38) {
						break ;
					}
					long msgId = buffToUInt(buff, 34);
					msgListnEventsList.offer(new MessageAcceptedEvent(new
							NodeId(buff, 18), msgId));
					break;
				}
				case PacketHeaders.RT_REQUEST:
				{
					if(recvPacket.getLength() != 24) {
						break ;
					}
					byte[] ipAddress		= Arrays.copyOfRange(buff, 18, 22);
					int port				= buffToUShort(buff, 22);
					processRTReq(new NodeId(buff, 2), ipAddress, port);
					break;
				}
				case PacketHeaders.RT_UPDATE:
				{
					if(recvPacket.getLength() < 358) {
						break ;
					}
					processRTUpdate(buff);
					if (isConnected == false) {
						++joinNumWait;
						if (joinNumWait == rtRespSize) {
							synchronized (isConnected) {
								conn = true;
								isConnected.notify();
							}
						}
					}
					break;
				}
				case PacketHeaders.ERROR:
				{
					if(recvPacket.getLength() < 20) {
						break ;
					}
					NodeId recvNodeId	= new NodeId(buff, 2);
					if (nodeId.compareTo(recvNodeId) != 0) {
						forwardError(recvNodeId, buff);
						break;
					}
					if(buffToUShort(buff, 18) == PacketHeaders.ERROR_NODE_NOT_FOUND) {
						msgListnEventsList.offer(new NodeNotFoundEvent(new
								NodeId(buff, 18)));
					}
					break;
				}
				default:
					System.err.println("Incompatible Header " + packetHeader
							+ "data: " + Arrays.toString(Arrays.copyOf(buff, recvPacket.getLength())));
					break;
				}
			}
			
		}
		
		private void forwardKSearch(NodeEntry nodeEntry, byte[] buff) {
			System.out.println("KSEARCH forward: "
					+ new NodeId(buff, 2) + " "
					+ nodeEntry.nodeId + " "
					+ nodeEntry.inetAddress
					+ " " + nodeEntry.port + "leafSet "
					+ leafSet + "\n");
			DatagramPacket packet = new DatagramPacket(buff, 26); 
			packet.setAddress(nodeEntry.inetAddress);
			packet.setPort(nodeEntry.port);
			synchronized (sendSocket) {
				try {
					sendSocket.send(packet);
				} catch (IOException e) {
					msgListnEventsList.offer(new IOErrorEvent(e));
				}
			}
		}
		
		private void forwardSendMessage(NodeId forwardId, byte[] buff, int length) {
			System.out.println("MESSAGE forward: " + forwardId);
			NodeEntry nodeEntry = nextHop(forwardId);
			if (nodeEntry == null) {
				processNodeNotFound(forwardId, new NodeId(buff, 18));
				return;
			}
			DatagramPacket packet = new DatagramPacket(buff, length); 
			packet.setAddress(nodeEntry.inetAddress);
			packet.setPort(nodeEntry.port);
			synchronized (sendSocket) {
				try {
					sendSocket.send(packet);
				} catch (IOException e) {
					msgListnEventsList.offer(new IOErrorEvent(e));
				}
			}
		}
		
		private void forwardError(NodeId forwardId, byte[] buff) {
			System.out.println("ERROR forward: " + forwardId);
			NodeEntry nodeEntry = nextHop(forwardId);
			if (nodeEntry == null) {
				return ;
			}
			DatagramPacket packet = new DatagramPacket(buff, 26); 
			packet.setAddress(nodeEntry.inetAddress);
			packet.setPort(nodeEntry.port);
			synchronized (sendSocket) {
				try {
					sendSocket.send(packet);
				} catch (IOException e) {
					msgListnEventsList.offer(new IOErrorEvent(e));
				}
			}
		}
		
		private void processKsearchResponse(NodeId kSearchId, byte[] addr, int port, int kSize) {
			InetAddress address = null;
			try {
				address = InetAddress.getByAddress(addr);
			} catch (UnknownHostException e) {
				e.printStackTrace();
			}
			processKSearchResponse(kSearchId, kSize, new InetSocketAddress(address, port));
		}
		
		private void processMessageAcceptResp(NodeId recvId, InetAddress address,
				int port, long msgId) {
			byte[] buff = new byte[38];
			uShortToBuff(buff, 0, PacketHeaders.MESSAGE_ACCEPTED);
			fillByteArrayBuff(buff, 2, recvId.getByteNodeId());
			fillByteArrayBuff(buff, 18, nodeId.getByteNodeId());
			uIntToBuff(buff, 34, msgId);
			DatagramPacket packet = new DatagramPacket(buff, buff.length); 
			packet.setAddress(address);
			packet.setPort(port);
			synchronized (sendSocket) {
				try {
					sendSocket.send(packet);
				} catch (IOException e) {
					msgListnEventsList.offer(new IOErrorEvent(e));
				}
			}
		}
		
		private void processRTReq(NodeId senderId, byte[] addr, int port) {
			InetAddress address = null;
			try {
				address = InetAddress.getByAddress(addr);
			} catch (UnknownHostException e) {
				e.printStackTrace();
			}
			processRTRequestResponse(new NodeEntry(senderId, address, port));
		}
		
		private void processRTUpdate(byte[] buff) {
			NodeEntry senderEntry = new NodeEntry();
			senderEntry.nodeId	= new NodeId(buff, 2);
			senderEntry.port	= buffToUShort(buff, 22);
			byte[] ip			= Arrays.copyOfRange(buff, 18, 22);
			try {
				senderEntry.inetAddress	= InetAddress.getByAddress(ip);
				routingTable.addNodeEntry(senderEntry);
				leafSet.addNodeEntry(senderEntry);
			} catch (UnknownHostException e) {
				e.printStackTrace();
			}
			for(int i = 0; i < 16; ++i) {
				NodeEntry entry = new NodeEntry();
				entry.nodeId	= new NodeId(buff, 24+i*22);
				if (entry.nodeId.compareTo(nodeId) == 0) {
					continue;
				}
				if (entry.nodeId.isNull()) {
					break;
				}
				byte[] ipAddress= Arrays.copyOfRange(buff, i*22+40, i*22+44);
				try {
					entry.inetAddress	= InetAddress.getByAddress(ipAddress);
				} catch (UnknownHostException e) {
					e.printStackTrace();
				}
				entry.port		= buffToUShort(buff, i*22+44);
				routingTable.addNodeEntry(entry);
				leafSet.addNodeEntry(entry);
			}
			long rtSize = buffToUInt(buff, 376);
			for(int i = 0; i < rtSize; ++i) {
				NodeEntry entry = new NodeEntry();
				entry.nodeId	= new NodeId(buff, 380+i*22);
				if (entry.nodeId.compareTo(nodeId) == 0) {
					continue;
				}
				if (entry.nodeId.isNull()) {
					break;
				}
				byte[] ipAddress= Arrays.copyOfRange(buff, i*22+396, i*22+400);
				try {
					entry.inetAddress	= InetAddress.getByAddress(ipAddress);
				} catch (UnknownHostException e) {
					e.printStackTrace();
				}
				entry.port		= buffToUShort(buff, i*22+400);
				routingTable.addNodeEntry(entry);
			}
		}
		
		private void processMsgWaitUpdate() {
			if (System.currentTimeMillis() - rtRequestTime > rtRequestSendTime) {
				processRTRequest();
			}
			synchronized (msgTimeOutMap) {
				Iterator<Entry<Long, MessageWaitEntry>> iter = msgTimeOutMap.entrySet().iterator();
				while (iter.hasNext()) {
					MessageWaitEntry waitEntry = iter.next().getValue();
					if (waitEntry.checkExpired()) {
						msgListnEventsList.offer(new MessageTimeOutEvent(waitEntry.msg));
						iter.remove();
					}
				}				
			}		
		}
	}
	
	public int getPort() {
		return port;
	}
	
	public NodeId getId() {
		return new NodeId(nodeId.getByteNodeId());
	}
	
}
